
package org.apache.ocean.name;

import java.io.File;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.net.URL;
import java.sql.Types;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.apache.commons.lang.StringUtils;
import org.apache.ocean.ClusterUtil;
import org.apache.ocean.FieldUtil;
import org.apache.ocean.NamedList;
import org.apache.ocean.TransactionID;
import org.apache.ocean.database.JDBCDatabase;
import org.apache.ocean.database.SQLDefaultParameters;
import org.apache.ocean.database.SQLParameters;
import org.apache.ocean.database.SQLResults;
import org.apache.ocean.database.SQLRow;
import org.apache.ocean.util.FileSequence;

/**
 *
 * @author jasonr
 */
public class SQLNameDatabase extends JDBCDatabase {
  public static final String NAME_TRANSACTION_ID_SEQUENCE_NAME = "nametransactionidsequence";
  public static Logger log = Logger.getLogger(SQLNameDatabase.class.getName());
  public NodeTransform nodeTransform = new NodeTransform();
  public NameService nameService;
  public static NameDefaults NAME_DEFAULTS = new NameDefaults();
  public FileSequence transactionIDSequence;
  
  static {
    System.setProperty("derby.storage.pageCacheSize", "2000");
  }
  
  public SQLNameDatabase(File directory) throws Exception {
    super("name", directory);
    transactionIDSequence = new FileSequence(directory.getAbsolutePath());
    transactionIDSequence.create(NAME_TRANSACTION_ID_SEQUENCE_NAME, new BigInteger("0"));
    nameService = new NameService();
    createTable("node", Node.class);
    syncColumns("node");
  }
  
  public TransactionID findHighestReplicationID() throws Exception {
  	List<String> tableNames = getTableNames();
  	List<TransactionID> list = new ArrayList<TransactionID>();
  	for (String tableName : tableNames) {
  		TransactionID id = findHighestReplicationID(tableName);
  		if (id != null) list.add(id);
  	}
  	return ClusterUtil.max(list);
  }
  
  public TransactionID findHighestReplicationID(String tableName) throws Exception {
  	String sql = "SELECT MAX(replicationID) FROM "+tableName;
    return executeQuery(sql, tableName).getAggregateTransactionID();
  }
  
  public boolean nodeExistsByID(String id) throws Exception {
  	return rowExists("id", id, "node");
  }
  
  protected void notifyUpdate() {
    try {
      transactionIDSequence.nextSequenceValueBottom(NAME_TRANSACTION_ID_SEQUENCE_NAME, 1);
    } catch (Throwable throwable) {
      log.log(Level.SEVERE, "", throwable);
    }
  }
  
  public URL getURL(String columnName, NamedList row) throws Exception {
    String string = (String)row.get(columnName);
    if (StringUtils.isBlank(string)) return null;
    return new URL(string);
  }
  
  public static class NameDefaults extends SQLDefaultParameters {
    public void setValues(String action, Object object) throws Exception {
      if (StringUtils.equals("update", action)) {
        FieldUtil.setFieldValue(  object,"dateModified", new java.util.Date());
      } else if (StringUtils.equals("insert", action)) {
        java.util.Date date = new java.util.Date();
        FieldUtil.setFieldValue(object,"dateModified", date);
        FieldUtil.setFieldValue(object,"dateCreated", date);
      }
    }
  }
  
  public TransactionID getTransactionID(String columnName, NamedList row) throws Exception {
    BigDecimal value = (BigDecimal)row.get(columnName);
    if (value == null) return null;
    return new TransactionID(value.toString());
  }
  
  public Boolean toBoolean(String value) {
    if (StringUtils.equals(value, "true")) return new Boolean(true);
    else return new Boolean(false);
  }
  
  public boolean hasNodeChanged(Node node) throws Exception {
    if (node.id == null) return false;
    Node otherNode = findNode(node.id);
    return !otherNode.equals(node);
  }
  
  public Node findNode(String id) throws Exception {
    String sql = "SELECT * FROM node WHERE id = ?";
    SQLResults results = executeQuery(sql, new SQLParameters(id, Types.VARCHAR), "node");
    return toNode(results.getFirstRow());
  }
  
  public Set<String> findDistinctCellIDs() throws Exception {
  	String sql = "SELECT DISTINCT(assignedCellID) FROM node";
  	SQLResults results = executeQuery(sql, null, "node");
  	log.info("results.size: "+results.size());
  	Set<String> cellIDs = new HashSet<String>(results.size());
  	for (SQLRow row : results) {
  		if (row.size() == 0) throw new Exception("row size 0: "+row);
  		String cellID = (String)row.get(0);
  		if (cellID != null) {
  		  cellIDs.add(cellID);
  		}
  	}
  	return cellIDs;
  }
  
  public List<Node> findNodesBeforeLastNodeCheck(java.util.Date date, String role) throws Exception {
    String sql = "SELECT * FROM node WHERE ? > lastNodeCheck AND role = ?";
    SQLParameters parameters = new SQLParameters(date, Types.TIMESTAMP, role, Types.VARCHAR);
    return executeQuery(sql, parameters, "node").toList(nodeTransform);
  }
  /**
   * public List<Node> findNodesBeforeLastNodeCheck(java.util.Date date) throws Exception {
   * String sql = "SELECT * FROM node WHERE ? > lastNodeCheck";
   * SQLResults results = executeQuery(sql, new SQLParameters(date, Types.TIMESTAMP), "node");
   * return results.toList(nodeTransform);
   * }
   **/
  public class NodeTransform implements SQLResults.Transform<Node> {
    public Node row(SQLRow row) throws Exception {
      return toNode(row);
    }
  }
  /**
   * public List<Node> findNodesByStatus(String status) throws Exception {
   * String sql = "SELECT * FROM node WHERE status = ? AND (role = ? OR role = ?) ";
   * SQLParameters parameters = new SQLParameters();
   * parameters.add(new SQLParameter(status, Types.VARCHAR));
   * parameters.add(new SQLParameter(ClusterConstants.MASTER, Types.VARCHAR));
   * parameters.add(new SQLParameter(ClusterConstants.SLAVE, Types.VARCHAR));
   * List<NamedList<String,Object>> results = executeQuery(sql, parameters, "node");
   * List<Node> nodes = new ArrayList<Node>();
   * for (NamedList row : results) {
   * Node node = toNode(row);
   * nodes.add(node);
   * }
   * return nodes;
   * }
   **/
  /**
   * public List<Node> findNodesNullCellIDNotClient() throws Exception {
   * String sql = "SELECT * FROM node WHERE cellID IS NULL AND (NOT role = ? OR role IS NULL)";
   * List<SQLParameter> parameters = new ArrayList<SQLParameter>();
   * parameters.add(new SQLParameter(ClusterConstants.CLIENT, Types.VARCHAR));
   * List<NamedList<String,Object>> results = executeQuery(sql, parameters, "node");
   * List<Node> nodes = new ArrayList<Node>();
   * for (NamedList row : results) {
   * Node node = toNode(row);
   * nodes.add(node);
   * }
   * return nodes;
   * }
   **/
  public List<Node> findNodesByAssignedCellID(String cellID) throws Exception {
    String sql = "SELECT * FROM node WHERE assignedCellID = ?";
    return executeQuery(sql, new SQLParameters(cellID, Types.VARCHAR), "node").toList(nodeTransform);
  }
  /**
   * public List<NodeShard> findShardsByCellID(String cellID) throws Exception {
   * String sql = "SELECT * FROM shard WHERE cellID = ?";
   * List<SQLParameter> parameters = new ArrayList<SQLParameter>(1);
   * parameters.add(new SQLParameter(cellID, Types.VARCHAR));
   * List<NamedList<String,Object>> results = executeQuery(sql, parameters, "cell");
   * List<NodeShard> nodeShards = new ArrayList<NodeShard>(results.size());
   * for (NamedList<String,Object> row : results) {
   * NodeShard nodeShard = toNodeShard(row);
   * nodeShards.add(nodeShard);
   * }
   * return nodeShards;
   * }
   **/
  
  /**
   * public Cell toCell(SQLRow row) throws Exception {
   * return (Cell)rowToObject(row, Cell.class, NAME_DEFAULTS);
   * }
   **/
  public Node toNode(SQLRow row) throws Exception {
    return (Node)rowToObject(row, Node.class, NAME_DEFAULTS);
  }
  /**
   * public NodeShard toNodeShard(NamedList<String,Object> row) throws Exception {
   * return (NodeShard)rowToObject(row, NodeShard.class, NAME_DEFAULTS);
   * }
   **/
  /**
   * public Cell findCell(String id) throws Exception {
   * String sql = "SELECT * FROM cell WHERE id = ?";
   * SQLResults results = executeQuery(sql, new SQLParameters(id, Types.VARCHAR), "cell");
   * return toCell(results.getFirstRow());
   * }
   *
   * public List<Cell> findCells() throws Exception {
   * return findCells("");
   * }
   *
   * public List<Cell> findCells(String orderBy) throws Exception {
   * return findCells(orderBy);
   * }
   **/
  /**
   * public List<Node> findNodesByRole(String role) throws Exception {
   * String sql = "SELECT * FROM node WHERE role = ?";
   * List<SQLParameter> parameters = new ArrayList<SQLParameter>(1);
   * parameters.add(new SQLParameter(role, Types.VARCHAR));
   * List<NamedList<String,Object>> results = executeQuery(sql, parameters, "node");
   * return toNodes(results);
   * }
   */
  /**
   * public List<Cell> findCellsAll() throws Exception {
   * return findCellsAll("");
   * }
   *
   * public List<Cell> findCellsAll(String orderBy) throws Exception {
   * String sql = "SELECT * FROM cell "+orderBy;
   * List<NamedList<String,Object>> results = executeQuery(sql, "cell");
   * List<Cell> cells = new ArrayList<Cell>();
   * for (NamedList<String,Object> row : results) {
   * Cell cell = toCell(row);
   * cells.add(cell);
   * }
   * return cells;
   * }
   **/
  public int findNodeCountByCellID(String cellID) throws Exception {
    if (StringUtils.isBlank(cellID)) throw new IllegalArgumentException("cellID null");
    String sql = "SELECT COUNT(id) FROM node WHERE cellID = ?";
    SQLResults results = executeQuery(sql, new SQLParameters(cellID, Types.VARCHAR), "node");
    return results.getCount();
  }
  
  /**
   * public Node findNodeByRoleAndCellID(String role, String cellID) throws Exception {
   * String sql = "SELECT * FROM node WHERE role = ? AND cellID = ?";
   * List<SQLParameter> parameters = new ArrayList<SQLParameter>(1);
   * parameters.add(getSQLParameter("role", role, "node"));
   * parameters.add(getSQLParameter("cellID", cellID, "node"));
   * List<NamedList<String,Object>> results = executeQuery(sql, parameters, "node");
   * NamedList<String,Object> row = getSingleRow(results);
   * if (row != null) return toNode(row);
   * return null;
   * }
   **/
  /**
   * public List<Node> findNodesNotRole(String orderBy, String role) throws Exception {
   * String sql = "SELECT * FROM node ORDER BY "+orderBy;
   * List<SQLParameter> parameters = new ArrayList<SQLParameter>(1);
   * //parameters.add(new Parameter(role, Types.VARCHAR));
   * List<NamedList<String,Object>> results = executeQuery(sql, parameters, "node");
   * List<Node> nodes = toNodes(results);
   * List<Node> nodeResults = new ArrayList<Node>();
   * for (Node node : nodes) {
   * if (!StringUtils.equals(node.role, role)) {
   * nodeResults.add(node);
   * }
   * }
   * return nodeResults;
   * }
   **/
  public List<Node> findNodes() throws Exception {
    String sql = "SELECT * FROM node";
    return executeQuery(sql, "node").toList(nodeTransform);
  }
  
  public List<Node> findNodes(String orderBy) throws Exception {
    String sql = "SELECT * FROM node ORDER BY "+orderBy;
    return executeQuery(sql, "node").toList(nodeTransform);
  }
  /**
   * public void deleteCell(String id) throws Exception {
   * String sql = "DELETE FROM cell WHERE id = ?";
   * List<SQLParameter> parameters = new ArrayList<SQLParameter>();
   * parameters.add(new SQLParameter(id, Types.VARCHAR));
   * executeUpdate(sql, parameters, "cell");
   * }
   *
   * public void deleteNodesByCellID(String cellID) throws Exception {
   * String sql = "DELETE FROM node WHERE cellid = ?";
   * List<SQLParameter> parameters = new ArrayList<SQLParameter>();
   * parameters.add(new SQLParameter(cellID, Types.VARCHAR));
   * executeUpdate(sql, parameters, "node");
   * }
   *
   * public void deleteNode(String id) throws Exception {
   * String sql = "DELETE FROM node WHERE id = ?";
   * List<SQLParameter> parameters = new ArrayList<SQLParameter>(1);
   * parameters.add(new SQLParameter(id, Types.VARCHAR));
   * executeUpdate(sql, parameters, "node");
   * }
   ***/
  /**
   * public CellShard updateCellShardSet(CellShard cellShard, boolean returnResponse) throws Exception {
   * cellShard.id = cellShard.getID();
   * return (CellShard)updateSet("cellshard", "id", cellShard, returnResponse, NAME_DEFAULTS);
   * }
   *
   * public Node updateNodeSet(Node node, boolean returnResponse) throws Exception {
   * return (Node)updateSet("node", "id", node, returnResponse, NAME_DEFAULTS);
   * //return updateSet("node", "id", node, returnResponse);
   * }
   **/
  public Node updateNode(Node node) throws Exception {
    //if (node.highestTransactionID == null) {
    //  node.highestTransactionID = new TransactionID("0");
    //}
    //if (node.highestTransactionLogID == null) {
    //  node.highestTransactionLogID = new TransactionID("0");
    //}
    return (Node)updateRow("node", node, NAME_DEFAULTS);
  }
  
  public Node updateNodeSet(Map<String,Object> map, boolean doReturn) throws Exception {
    return (Node)updateSet("node", map, Node.class, doReturn, NAME_DEFAULTS);
  }
  
  public Node updateNodeSet(Node node, boolean doReturn) throws Exception {
    return (Node)updateSet("node", node, doReturn, NAME_DEFAULTS);
  }
  
  public String toString(Boolean value) {
    if (value == null) return null;
    else return value.toString();
  }
  /**
   * public CellShard updateCellShard(CellShard cellShard) throws Exception {
   * return (CellShard)updateStandard("cellshard", "id", cellShard, NAME_DEFAULTS);
   * }
   *
   * public NodeShard updateNodeShardSet(NodeShard shard, boolean returnResponse) throws Exception {
   * return (NodeShard)updateSet("shard", "id", shard, returnResponse, NAME_DEFAULTS);
   * }
   *
   * public NodeShard updateNodeShard(NodeShard shard) throws Exception {
   * return (NodeShard)updateStandard("shard", "id", shard, NAME_DEFAULTS);
   * }
   **/
  //public void updateCell(Cell cell) throws Exception {
  //  updateStandard("cell", "id", cell, NAME_DEFAULTS);
  //}
  /**
   * public void createTables() {
   * executeUpdateIgnoreException(createCellSQL);
   * addParameterTypeMap("cell", createCellParameterTypes());
   *
   * executeUpdateIgnoreException(createNodeSQL);
   * executeUpdateIgnoreException(createNodeCellIndexSQL);
   * executeUpdateIgnoreException(createNodeStatusIndexSQL);
   * executeUpdateIgnoreException(createNodeRoleIndexSQL);
   * addParameterTypeMap("node", createNodeParameterTypes());
   *
   * executeUpdateIgnoreException(createShardSQL);
   * executeUpdateIgnoreException(createShardStatusIndexSQL);
   * executeUpdateIgnoreException(createShardRoleIndexSQL);
   * executeUpdateIgnoreException(createShardCellIndexSQL);
   * executeUpdateIgnoreException(createShardNodeIndexSQL);
   * addParameterTypeMap("shard", createShardParameterTypes());
   *
   * executeUpdateIgnoreException(createCellShardSQL);
   * executeUpdateIgnoreException(createCellShardShardIDIndexSQL);
   * executeUpdateIgnoreException(createCellShardCellIDIndexSQL);
   * addParameterTypeMap("cellshard", createCellShardParameterTypes());
   *
   * //executeUpdateIgnoreException(createShardHistorySQL);
   * //executeUpdateIgnoreException(createShardHistoryShardIDIndexSQL);
   * //executeUpdateIgnoreException(createShardHistoryCellIndexSQL);
   * }
   **/
  /**
   * public static String createNodeSQL =
   * "CREATE TABLE node " +
   * "(id VARCHAR(255) PRIMARY KEY, " +
   * "assignedCellID VARCHAR(255)," +
   * "assignedStatus VARCHAR(255),"+
   * "reportedCellID VARCHAR(255),"+
   * "reportedStatus VARCHAR(255),"+
   * "reportedMasterID VARCHAR(255),"+
   * "reportedMasterIDSinceDate TIMESTAMP,"+
   * "dateReported TIMESTAMP,"+
   * "lastLeaseRenewalDate TIMESTAMP,"+
   * "highestTransactionIDLog NUMERIC," +
   * "lowestTransactionIDLog NUMERIC," +
   * "highestTransactionIDIndex NUMERIC," +
   * "lowestTransactionIDIndex NUMERIC," +
   * "url VARCHAR(999),"+
   * "dateModified TIMESTAMP," +
   * "dateCreated TIMESTAMP,"+
   * "totalSize BIGINT,"+
   * "totalIndexSize BIGINT,"+
   * "totalTransactionLogSize BIGINT"+
   * ")";
   *
   * public static String createNodeCellIndexSQL =
   * "CREATE INDEX node_cellid_idx ON node(cellID)";
   * public static String createNodeStatusIndexSQL =
   * "CREATE INDEX node_status_idx ON node(status)";
   * public static String createNodeRoleIndexSQL =
   * "CREATE INDEX node_role_idx ON node(role)";
   *
   * public SQLTable createNodeTable() {
   * SQLTable table = new SQLTable("node");
   * table.addCreateTable(createNodeSQL);
   * table.addPostCreateTableSQL(createNodeCellIndexSQL);
   * table.addPostCreateTableSQL(createNodeStatusIndexSQL);
   * table.addPostCreateTableSQL(createNodeRoleIndexSQL);
   * table.add("id", Types.VARCHAR);
   * table.add("assignedCellID", Types.VARCHAR);
   * table.add("assignedStatus", Types.VARCHAR);
   * table.add("reportedCellID", Types.VARCHAR);
   * table.add("reportedStatus", Types.VARCHAR);
   * table.add("reportedMasterID", Types.VARCHAR);
   * table.add("reportedMasterIDSinceDate", Types.TIMESTAMP);
   * table.add("dateReported", Types.TIMESTAMP);
   * table.add("lastLeaseRenewalDate", Types.TIMESTAMP);
   * table.add("highestTransactionIDLog", Types.NUMERIC);
   * table.add("lowestTransactionIDLog", Types.NUMERIC);
   * table.add("highestTransactionIDIndex", Types.NUMERIC);
   * table.add("lowestTransactionIDIndex", Types.NUMERIC);
   * table.add("url", Types.VARCHAR);
   * table.add("dateModified", Types.TIMESTAMP);
   * table.add("dateCreated", Types.TIMESTAMP);
   * table.add("totalSize", Types.BIGINT);
   * table.add("totalIndexSize", Types.BIGINT);
   * table.add("totalTransactionLogSize", Types.BIGINT);
   * return table;
   * }
   **/
  /**
   * public static String createCellShardShardIDIndexSQL =
   * "CREATE INDEX cellshard_shardid_idx ON cellshard(shardID)";
   * public static String createCellShardCellIDIndexSQL =
   * "CREATE INDEX cellshard_cellid_idx ON cellshard(cellID)";
   *
   * public static String createCellShardSQL =
   * "CREATE TABLE cellshard " +
   * "(id VARCHAR(255) PRIMARY KEY, " +
   * "shardID VARCHAR(255), "+
   * "cellID VARCHAR(255), "+
   * "status VARCHAR(255), "+
   * "assignedStatus VARCHAR(255), "+
   * "maxSize BIGINT, "+
   * "moveStartDate TIMESTAMP, " +
   * "moveStatus VARCHAR(255), "+
   * "movingNameID VARCHAR(255), "+
   * "sourceCellID VARCHAR(255), "+
   * "destinationCellID VARCHAR(255), "+
   * "dateModified TIMESTAMP, " +
   * "dateCreated TIMESTAMP)";
   *
   * public static Map<String,Integer> createCellShardParameterTypes() {
   * Map<String,Integer> map = new HashMap<String,Integer>();
   * add("id", Types.VARCHAR, map);
   * add("shardID", Types.VARCHAR, map);
   * add("cellID", Types.VARCHAR, map);
   * add("status", Types.VARCHAR, map);
   * add("assignedStatus", Types.VARCHAR, map);
   * add("maxSize", Types.BIGINT, map);
   * add("moveStartDate", Types.TIMESTAMP, map);
   * add("moveStatus", Types.VARCHAR, map);
   * add("movingNameID", Types.VARCHAR, map);
   * add("sourceCellID", Types.VARCHAR, map);
   * add("destinationCellID", Types.VARCHAR, map);
   * add("dateModified", Types.TIMESTAMP, map);
   * add("dateCreated", Types.TIMESTAMP, map);
   * return map;
   * }
   *
   * public static String createShardSQL =
   * "CREATE TABLE shard " +
   * "(id VARCHAR(255) PRIMARY KEY, " +
   * "cellID VARCHAR(255),"+
   * "nodeID VARCHAR(255),"+
   * "highestTransactionID NUMERIC, "+
   * "lowestTransactionID NUMERIC, " +
   * "highestTransactionLogID NUMERIC, " +
   * "lastCheck TIMESTAMP, "+
   * "assignedStatus VARCHAR(255), "+
   * "status VARCHAR(255), "+
   * "moveStatus VARCHAR(255), "+
   * "healthCheck VARCHAR(999), "+
   * "isReloading VARCHAR(999), "+
   * "dateModified TIMESTAMP, " +
   * "dateCreated TIMESTAMP,"+
   * "size BIGINT, "+
   * "indexSize BIGINT, "+
   * "transactionLogSize BIGINT"+
   * ")";
   *
   * public static String createShardNodeIndexSQL =
   * "CREATE INDEX shard_nodeid_idx ON shard(nodeID)";
   * public static String createShardCellIndexSQL =
   * "CREATE INDEX shard_cellid_idx ON shard(cellID)";
   * public static String createShardStatusIndexSQL =
   * "CREATE INDEX shard_status_idx ON shard(status)";
   * public static String createShardRoleIndexSQL =
   * "CREATE INDEX shard_role_idx ON shard(role)";
   *
   * public static Map<String,Integer> createShardParameterTypes() {
   * Map<String,Integer> map = new HashMap<String,Integer>();
   * add("id", Types.VARCHAR, map);
   * add("cellID", Types.VARCHAR, map);
   * add("highestTransactionID", Types.NUMERIC, map);
   * add("lowestTransactionID", Types.NUMERIC, map);
   * add("highestTransactionLogID", Types.NUMERIC, map);
   * add("lastCheck", Types.TIMESTAMP, map);
   * add("assignedStatus", Types.VARCHAR, map);
   * add("status", Types.VARCHAR, map);
   * add("moveStatus", Types.VARCHAR, map);
   * add("healthCheck", Types.VARCHAR, map);
   * add("isReloading", Types.VARCHAR, map);
   * add("dateModified", Types.TIMESTAMP, map);
   * add("dateCreated", Types.TIMESTAMP, map);
   * return map;
   * }
   **/
  
  /**
   * public static String createShardHistorySQL =
   * "CREATE TABLE shardHistory (id VARCHAR(255) PRIMARY KEY, " +
   * "shardID VARCHAR(255),"+
   * "cellID VARCHAR(255),"+
   * "status VARCHAR(255),"+
   * "dateModified TIMESTAMP, "+
   * "dateCreated TIMESTAMP"+
   * ")";
   * public static String createShardHistoryShardIDIndexSQL =
   * "CREATE INDEX shardhistory_shardid_idx ON shardHistory(shardID)";
   * public static String createShardHistoryCellIndexSQL =
   * "CREATE INDEX shardhistory_cellid_idx ON shardHistory(cellID)";
   **/
  
}
